package com.techcity;

import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.hadoop.io.IntWritable;

public class Maxv extends UDAF {
    public static class MaxState {
        private IntWritable result1;
        private IntWritable result2;
        private IntWritable result3;
    }
    public static class MaxEvaluator implements UDAFEvaluator {
        MaxState state;
        public MaxEvaluator(){
            super();
            state = new MaxState();
            init();
        }
        //负责初始化计算函数并重设它的内部状态。
        public void init() {
//            System.out.println("Init-1");
            state.result1 = null;
            state.result2 = null;
            state.result3 = null;
        }

        // 每次对一个新值进行聚集计算都会调用该方法，并进行内部的轮转。其中 iterate()接收的参数跟 hive 的参数对应。
        public boolean iterate(IntWritable value1,IntWritable value2,IntWritable value3) {
//            System.out.println("iterate-2");
            if (value1 == null && value2 == null && value3 == null) {
                return true;
            }
            if (state.result1 == null && state.result2 == null && state.result3 == null) {
                state.result1 = new IntWritable( value1.get() );
                state.result2 = new IntWritable( value2.get() );
                state.result3 = new IntWritable( value3.get() );
            } else {
                state.result1.set( Math.max( state.result1.get(), value1.get() ) );
                state.result2.set( Math.max( state.result2.get(), value2.get() ) );
                state.result3.set( Math.max( state.result3.get(), value3.get() ) );
            }
            return true;
        }
        // 部分聚集结果（iterate 函数轮转结束后的返回结果），返回一个封装了聚集计算当前状态的对象。terminatePartial 类似于 hadoop 的 Combiner。
        public MaxState terminatePartial() {
//            System.out.println("terminatePartial-3");

            return state;
        }

        // 接收 terminatePartial 的返回结果，进行数据 merge 操作，其返回类型为 boolean。
        public boolean merge(MaxState other) {
//            System.out.println("merge-4");

            return iterate( other.result1, other.result2, other.result3);
        }

        // 返回最终的聚集结果
        public String terminate() {
//            System.out.println("terminate-5");
            return state.result1+" "+state.result2+" "+state.result3;
        }
    }
}

//待改进